package edu.hit.pact.pgse.crawler.send;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IoSession;

import edu.hit.pact.pgse.bean.RawWebPage;
import edu.hit.pact.pgse.bean.RunnableTask;
import edu.hit.pact.pgse.bean.TaskPieceProperty;
import edu.hit.pact.pgse.communication.CommUtil;
import edu.hit.pact.pgse.communication.message.CrawlerStorePushMessage;
import edu.hit.pact.pgse.communication.message.Types;
import edu.hit.pact.pgse.crawler.web.PageFetchProcessor;
import edu.hit.pact.pgse.crawler.finish.FinishChecker;
import edu.hit.pact.pgse.crawler.util.SleepThread;
import edu.hit.pact.pgse.util.AbstractThread;
import edu.hit.pact.pgse.util.Globals;

public class RawPageSendProcessor extends AbstractThread {
	private Log logger;
	private int sendPageNumber = 0;
	private LinkedBlockingQueue<RawWebPage> fetchedPageQueue;
//	private Timer queueInfoRecorder;
	private TaskPieceProperty pieceProperty;
	public static IoSession session;
	private boolean ifSend;
	private ArrayList<RawWebPage> pagesToSend;
	private CommUtil communicationUtil;
//	private DSInfoOutput dsoutput;

	public RawPageSendProcessor(
			LinkedBlockingQueue<RawWebPage> fetchedPageQueue, RunnableTask runnableTask, boolean ifSend) {
		this(LogFactory.getLog(PageFetchProcessor.class),
				"PageSendProcessor", fetchedPageQueue, runnableTask, ifSend);
	}

	public RawPageSendProcessor(Log logger, String threadName,
			LinkedBlockingQueue<RawWebPage> fetchedPageQueue, RunnableTask runnableTask, boolean ifSend) {
		super(logger);
		this.logger = logger;
		this.fetchedPageQueue = fetchedPageQueue;
//		this.queueInfoRecorder = new Timer();
		this.pieceProperty = runnableTask.getTaskPieceProperty();
		this.ifSend = ifSend;
		this.pagesToSend = new ArrayList<RawWebPage>();
		
		if (ifSend){
			this.communicationUtil = new CommUtil(new RawPageSendHandler());
			this.session = communicationUtil.setConnector(runnableTask.getStoreAddr());
		} else {
			this.session = null;
		}
//		dsoutput = new DSInfoOutput();
//		queueInfoRecorder.schedule(dsoutput, 1 * 1000, 1 * 1000);
	}

	@Override
	public void mainRun() {
		SleepThread.sleep(Globals.CRAWLER_SEND_RAWPAGE_INTERVAL);	
		pagesToSend.clear();
		System.gc();
		
		//when the piece is finished, this page send processor waits 10 seconds, then closes
		if (FinishChecker.isFinish(logger, pieceProperty) && fetchedPageQueue.size() == 0) {
			logger.error("all pages were sent out, page send processor is dying!");
//			dsoutput.setDeadLoop(false);
			setDead();
			return;
		}
		
		try {
			//不足20个就一直等
			while (pagesToSend.size() < Globals.CRAWLER_SEND_RAWPAGE_ONCE_PAGENUMBER){
				if (FinishChecker.isFinish(logger, pieceProperty) && fetchedPageQueue.size() == 0){
					logger.info("piece is finished & fetch queue is zero, but still have pages to send");
					break;
				} else if (fetchedPageQueue.size() == 0 && !FinishChecker.isFinish(logger, pieceProperty)){
					logger.info("piece is not finished & fetch queue is zero");
					Thread.sleep(5000);
				} else if (fetchedPageQueue.size() >0){
					logger.info("remaining " + fetchedPageQueue.size() + " pages");
					pagesToSend.add(this.fetchedPageQueue.take());
				}
			}
			
			if (ifSend){
				if (session != null){//indicate that the session has been connected
					logger.info("send "+pagesToSend.size()+" new pages");
					session.write(new CrawlerStorePushMessage(Types.CRAWEL_STORE_PUSH, pagesToSend));
				}
				else{
					logger.error("session is null");
				}
			} 
			
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		sendPageNumber += pagesToSend.size();
		logger.error("send "+pagesToSend.size()+", remain "+this.fetchedPageQueue.size()+". Have send "+ sendPageNumber + " pages.");
	
	}

	public void setDead() {
		fetchedPageQueue = null;
		setDeadLoop(false);
		communicationUtil.closeConnector();
	}
}

//	class DSerrorOutput extends TimerTask {
//		private PrintWriter queueTracker = null;
//		private boolean isDeadLoop = true;
//		private SimpleDateFormat formater = new SimpleDateFormat(
//				"yyyy-MM-dd*hh:mm:ss");
//
//		public DSerrorOutput() {
//			try {
//				queueTracker = new PrintWriter(new FileOutputStream(
//						"PageSendProcessor.txt", true), true);
//				queueTracker.println("\t\t\ttotal\tfetched\tbuffer");
//			} catch (FileNotFoundException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
//		}
//
//		public void setDeadLoop(boolean loop) {
//			this.isDeadLoop = loop;
//		}
//
//		public void run() {
//			if (!isDeadLoop)
//				return;
//
//			queueTracker.println(formater.format(new Date()) + "\t"
//					+ sendPageNumber + "\t" + fetchedPageQueue.size());
//		}
//
//	}
//}
